package com.gitee.free.coding.demo.reactive;

import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

@Slf4j
public class ReactiveTest {

    public static void main(String[] args) {
        testReactiveStream();
    }

    /**
     * 测试响应式编程
     */
    private static void testReactiveStream() {
        // 发布者
        try (SubmissionPublisher<String> publisher = new SubmissionPublisher<>()) {
            // 发布消息
            publishMessage(publisher);

            Thread.sleep(20000);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 发布消息
     *
     * @param publisher 发布者
     */
    private static void publishMessage(SubmissionPublisher<String> publisher) {
        // 链路 ==>  发布者[发布消息] -> 处理器[处理并转发消息] -> 订阅者[处理消息]
        // 处理器。可以是发布者，也可以是订阅者
        Flow.Processor<String, String> processor = new StreamProcessor();
        // 发布者绑定处理器
        publisher.subscribe(processor);

        // 订阅者
        Flow.Subscriber<String> subscriber = getSubscriber();
        // 处理器绑定订阅者
        processor.subscribe(subscriber);

        // 发布消息
        for (int i = 0; i < 10; i++) {
            publisher.submit("message[" + i + "]");
        }
    }

    /**
     * 获取订阅者
     *
     * @return 订阅者
     */
    private static Flow.Subscriber<String> getSubscriber() {
        return new Flow.Subscriber<>() {

            private Flow.Subscription subscription;

            /**
             * 订阅时触发
             * 注意：必须保证 onSubscribe 方法在 onNext 方法前面执行，否则 this.subscription.request(1L); 会报空指针异常
             *      首先需要调用 [SubmissionPublisher] 类中的 subscribe 方法，也就是发布订阅绑定。如 publisher.subscribe(processor); 在该方法中会调用对应的 onSubscribe 方法
             *      如果是自定义的 Flow.Processor 类，那就要继承 SubmissionPublisher 类，重写 subscribe 方法，在 subscribe 方法中使用 super.subscribe(subscriber);
             *      确保绑定时订阅者就调用 onSubscribe 方法为 this.subscription 赋值
             * @param subscription a new subscription
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                log.info("===== getSubscriber onSubscribe =====");
                this.subscription = subscription;
                // 找上游要一个数据
                this.subscription.request(1L);
            }

            /**
             * 下一个元素到达时，执行该回调
             * @param item the item
             */
            @Override
            public void onNext(String item) {
                log.info("===== getSubscriber onNext =====");
                log.info("订阅者拿到数据：{}，处理......", item);
                if ("message[5]".equals(item)) {
                    // 退出，不再拿数据了
                    this.subscription.cancel();
                    return;
                }
                item += " - [Subscriber已处理]";
                // 找上游要一个数据
                this.subscription.request(1L);
                log.info("处理完成：{}", item);
            }

            /**
             * 发生错误时
             * @param throwable the exception
             */
            @Override
            public void onError(Throwable throwable) {
                log.info("===== getSubscriber onError =====");
                log.error(throwable.getMessage(), throwable);
            }

            /**
             * 完成时
             */
            @Override
            public void onComplete() {
                log.info("===== getSubscriber onComplete =====");
            }
        };
    }

    static class StreamProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {

        /**
         * 缓存订阅了当前 StreamProcessor 的所有订阅者
         */
        private final List<Flow.Subscriber<? super String>> subscribers = new CopyOnWriteArrayList<>();

        /**
         * 缓存当前 Flow.Subscription
         */
        private Flow.Subscription subscription;

        @Override
        public void subscribe(Flow.Subscriber<? super String> subscriber) {
            // 确保在这里先调用 subscriber 的 onSubscribe 方法使 this.subscription 赋值，否则会报空指针异常
            super.subscribe(subscriber);
            log.info("===== StreamProcessor subscribe =====");
            subscribers.add(subscriber);
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            log.info("===== StreamProcessor onSubscribe =====");
            this.subscription = subscription;
            // 找上游要一个数据
            this.subscription.request(1L);
        }

        @Override
        public void onNext(String item) {
            log.info("===== StreamProcessor onNext =====");
            // 找上游要一个数据
            this.subscription.request(1L);
            log.info("拿到了数据：{}，开始处理...", item);
            item += " - [StreamProcessor已处理]";
            log.info("处理完成：{}", item);

            // 向订阅了当前 Flow.Processor 的订阅者发送消息
            for (Flow.Subscriber<? super String> subscriber : subscribers) {
                subscriber.onNext(item);
            }
        }

        @Override
        public void onError(Throwable throwable) {
            log.info("===== StreamProcessor onError =====");
            log.error(throwable.getMessage(), throwable);
        }

        @Override
        public void onComplete() {
            log.info("===== StreamProcessor onComplete =====");
        }
    }

}
